package com.bw.gmall.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.bean.TrafficPageViewBean;
import com.bw.gmall.utils.MyClickHouseUtil;
import com.bw.gmall.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.time.Duration;

public class DwsTrafficVcChArIsNewPageViewWindow2 {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        String uvTopic = "dwd_traffic_unique_visitor_detail";
        String ujdTopic = "dwd_traffic_user_jump_detail";
        String Topic = "dwd_traffic_page_log";
        String groupID = "vccharisnew_pageview_window_2";

        DataStreamSource<String> uvDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(uvTopic, groupID));
        DataStreamSource<String> ujDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(ujdTopic, groupID));
        DataStreamSource<String> pageDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(Topic, groupID));

        SingleOutputStreamOperator<TrafficPageViewBean> traUvDS = (SingleOutputStreamOperator<TrafficPageViewBean>) uvDS.map(new MapFunction<String, TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean map(String value) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(value);
                JSONObject common = jsonObject.getJSONObject("common");
                String vc = common.getString("vc");
                String ch = common.getString("ch");
                String ar = common.getString("ar");
                String isNew = common.getString("is_new");
                return new TrafficPageViewBean(null, null,
                        vc, ch, ar, isNew,
                        1L, 0L, 0L, 0L, 0L,
                        jsonObject.getLong("ts"));
            }
        });

        SingleOutputStreamOperator<TrafficPageViewBean> traUjDS = ujDS.map(new MapFunction<String, TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean map(String value) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(value);
                JSONObject common = jsonObject.getJSONObject("common");
                String vc = common.getString("vc");
                String ch = common.getString("ch");
                String ar = common.getString("ar");
                String isNew = common.getString("is_new");
                return new TrafficPageViewBean(null, null,
                        vc, ch, ar, isNew,
                        0L, 0L, 0L, 0L, 1L,
                        jsonObject.getLong("ts"));
            }
        });

        SingleOutputStreamOperator<TrafficPageViewBean> TraPageDS = pageDS.map(new MapFunction<String, TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean map(String value) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(value);
                JSONObject common = jsonObject.getJSONObject("common");
                String vc = common.getString("vc");
                String ch = common.getString("ch");
                String ar = common.getString("ar");
                String isNew = common.getString("is_new");
                JSONObject page = jsonObject.getJSONObject("page");
                String last_page_id = page.getString("last_page_id");
                long sv = 0L;
                if (last_page_id == null) {
                    sv = 1L;
                }
                return new TrafficPageViewBean(null, null, vc, ch, ar, isNew,
                        0L, sv, 1L, page.getLong("during_time"), 0L, jsonObject.getLong("ts"));
            }
        });
        DataStream<TrafficPageViewBean> unionDS = traUvDS.union(traUjDS, TraPageDS);

        SingleOutputStreamOperator<TrafficPageViewBean> traPVBDS = unionDS.assignTimestampsAndWatermarks(WatermarkStrategy
                .<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner(new SerializableTimestampAssigner<TrafficPageViewBean>() {
                    @Override
                    public long extractTimestamp(TrafficPageViewBean element, long recordTimestamp) {
                        return element.getTs();
                    }
                }));

        KeyedStream<TrafficPageViewBean, Tuple4<String, String, String, String>> keyedStream = traPVBDS.keyBy(new KeySelector<TrafficPageViewBean, Tuple4<String, String, String, String>>() {
            @Override
            public Tuple4<String, String, String, String> getKey(TrafficPageViewBean value) throws Exception {

                return Tuple4.of(value.getVc(), value.getCh(), value.getAr(), value.getIsNew());
            }
        });

        SingleOutputStreamOperator<TrafficPageViewBean> aggregate = keyedStream
                .window(TumblingEventTimeWindows.of(Time.seconds(10L)))
                .allowedLateness(Time.seconds(10L))
                .aggregate(new AggregateFunction<TrafficPageViewBean, Tuple5<Long, Long, Long, Long, Long>, TrafficPageViewBean>() {
                    @Override
                    public Tuple5<Long, Long, Long, Long, Long> createAccumulator() {

                        return Tuple5.of(0L, 0L, 0L, 0L, 0L);
                    }

                    @Override
                    public Tuple5<Long, Long, Long, Long, Long> add(TrafficPageViewBean value,
                                                                    Tuple5<Long, Long, Long, Long, Long> accumulator) {

                        return Tuple5.of(value.getUvCt() + accumulator.f0,
                                value.getSvCt() + accumulator.f1,
                                value.getPvCt() + accumulator.f2,
                                value.getDurSum() + accumulator.f3,
                                value.getUjCt() + accumulator.f4);
                    }

                    @Override
                    public TrafficPageViewBean getResult(Tuple5<Long, Long, Long, Long, Long> accumulator) {
                        TrafficPageViewBean trafficPageViewBean = new TrafficPageViewBean();
                        trafficPageViewBean.setUvCt(accumulator.f0);
                        trafficPageViewBean.setSvCt(accumulator.f1);
                        trafficPageViewBean.setPvCt(accumulator.f2);
                        trafficPageViewBean.setDurSum(accumulator.f3);
                        trafficPageViewBean.setUjCt(accumulator.f4);
                        return trafficPageViewBean;
                    }

                    @Override
                    public Tuple5<Long, Long, Long, Long, Long> merge(Tuple5<Long, Long, Long, Long, Long> a, Tuple5<Long, Long, Long, Long, Long> b) {
                        return null;
                    }
                }, new WindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple4<String, String, String, String> stringStringStringStringTuple4,
                                      TimeWindow window,
                                      Iterable<TrafficPageViewBean> input,
                                      Collector<TrafficPageViewBean> out) throws Exception {
                        TrafficPageViewBean next = input.iterator().next();
                        String vc = stringStringStringStringTuple4.f0;
                        String ch = stringStringStringStringTuple4.f1;
                        String ar = stringStringStringStringTuple4.f2;
                        String isNew = stringStringStringStringTuple4.f3;
                        Timestamp start = new Timestamp(window.getStart());
                        Timestamp end = new Timestamp(window.getEnd());
                        Long ts = System.currentTimeMillis();
                        TrafficPageViewBean trafficPageViewBean = new TrafficPageViewBean(start,
                                end,
                                vc,
                                ch,
                                ar,
                                isNew,
                                next.getUvCt(),
                                next.getSvCt(), next.getPvCt(),
                                next.getDurSum(), next.getUjCt(), ts);
                    }
                });
        aggregate.print();
        aggregate.addSink(MyClickHouseUtil.getSinkFunction("insert into table gmall.dws_traffic_vc_ch_page_window values(?,?,?,?,?,?,?,?,?,?,?,?)"));



    }
}
